package cn.ac.iie.ulss.dataredistribution.handler;

import cn.ac.iie.ulss.dataredistribution.commons.GlobalVariables;
import cn.ac.iie.ulss.dataredistribution.commons.RuntimeEnv;
import cn.ac.iie.ulss.dataredistribution.tools.HNode;
import cn.ac.iie.ulss.dataredistribution.tools.INode;
import java.util.ArrayList;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.log4j.PropertyConfigurator;

/**
 *
 * @author: evan
 * @date: 2014-10-15
 */
class RemoveNodeFromMessageTransferStation implements Runnable {

    String table = null;
    ArrayList<HNode> nurl = null;
    static org.apache.log4j.Logger logger = null;

    static {
        PropertyConfigurator.configure("log4j.properties");
        logger = org.apache.log4j.Logger.getLogger(RemoveNodeFromMessageTransferStation.class.getName());
    }

    public RemoveNodeFromMessageTransferStation(String table, ArrayList<HNode> nurl) {
        this.table = table;
        this.nurl = nurl;
    }

    @Override
    public void run() {
        ConcurrentHashMap<String, ConcurrentHashMap<HNode, ConcurrentHashMap<INode, ConcurrentLinkedQueue>>> tableToMSGStation = (ConcurrentHashMap<String, ConcurrentHashMap<HNode, ConcurrentHashMap<INode, ConcurrentLinkedQueue>>>) RuntimeEnv.getParam(GlobalVariables.TABLE_TO_MSGSTATION);
        ConcurrentHashMap<HNode, ConcurrentHashMap<INode, ConcurrentLinkedQueue>> messageTransferStation = tableToMSGStation.get(table);
                
        for (HNode n : nurl) {
            ConcurrentHashMap<INode, ConcurrentLinkedQueue> chm = (ConcurrentHashMap<INode, ConcurrentLinkedQueue>) messageTransferStation.get(n);
            while (true) {
                synchronized (chm) {
                    if (chm.isEmpty()) {
                        messageTransferStation.remove(n);
                        logger.info("messageTransmitStation remove the node " + n.getName() + " for the " + table);
                        break;
                    } else {
                        try {
                            Thread.sleep(5000);
                        } catch (InterruptedException ex) {
                            logger.error(ex, ex);
                        }
                    }
                }
            }
        }
    }
}
